// External Merge Sort Algorithm modified by LU JINBO@2010-10-10

package qp.operators;

import java.io.*;
import java.util.*;
import qp.utils.*;
import qp.optimizer.*;
import qp.parser.*;
import qp.operators.*;

public class ExternalMergeSort
{
	static int IOCost=0;
	static int stage=0;

	Schema SchemaForSort; // contain the index of tuple
	static int TempNum = 0;
	LinkedList<String> FileInSorting;

	ObjectInputStream in;
	ObjectOutputStream out;

	int BufferNumForsort;// buffer size used for sorting, assigned by buffer
							// manager
	int batchSize;
	int tupleSize;

	Vector<Attribute> AttriVector;
	Vector<Integer> AttriIndex = new Vector<Integer>();

	public ExternalMergeSort(String ToSortedFile, Vector<Attribute> Attibutions, Schema ToSortedFileSchema)
	{
		SchemaForSort = ToSortedFileSchema;

		try
		{
			in = new ObjectInputStream(new FileInputStream(ToSortedFile));
		} catch (Exception io)
		{
			System.err.println("Error");
		}

		AttriVector = Attibutions;
		for (int i = 0; i < AttriVector.size(); i++)
		{
			int index = SchemaForSort.indexOf(AttriVector.elementAt(i));
			AttriIndex.add(index);
		}

		BufferNumForsort = BufferManager.getBufferForSort();

		tupleSize = SchemaForSort.getTupleSize();
		batchSize = Batch.getPageSize() / tupleSize;
		FileInSorting = new LinkedList<String>();
	}

	private String GenerateFile()
	{
		TempNum++;
		return (String.valueOf(TempNum) + ".tmp");
	}

	public String MergeSort() throws IOException, ClassNotFoundException
	{

		Block BlockForRead = new Block(BufferNumForsort - 1);
		boolean eof = false;

		Vector<String> FileToRead = new Vector<String>();
		Vector<String> FileToWrite = new Vector<String>();

		Batch outBatch = new Batch(batchSize);
		Vector<ObjectInputStream> blockois;
		String outFile = null;
		ObjectOutputStream oos = null;

		int min = 0;

		while (!eof)
		{
			BlockForRead = new Block(BufferNumForsort - 1);
			outFile = GenerateFile();

			for (int i = 0; i < BufferNumForsort - 1; i++)
			{
				try
				{
					BlockForRead.add(i, (Batch) in.readObject());
					IOCost++;
				} catch (Exception e)
				{
					eof = true;
					break;
				}
			}

			if (BlockForRead.size() > 0)
			{
				for (int i = 0; i < BlockForRead.size(); i++)
				{
					BlockForRead.batchAt(i).sort(AttriIndex.firstElement());
				}
				try
				{
					FileOutputStream fos = new FileOutputStream(outFile);
					oos = new ObjectOutputStream(fos);
					while (BlockForRead.existTupleInBlock > 0)
					{
						min = BlockForRead.findMinFromSortedBatch(AttriIndex.firstElement());
						outBatch.add(BlockForRead.batchAt(min).elementAt(0));
						BlockForRead.batchAt(min).remove(0);
						BlockForRead.existTupleInBlock--;
						if (outBatch.isFull())
						{
							oos.writeObject(outBatch);
							IOCost++;
							outBatch = new Batch(batchSize);
						}
					}
					if (outBatch.size() > 0)
					{
						oos.writeObject(outBatch);
						IOCost++;
						outBatch = new Batch(batchSize);
					}
					oos.close();
					fos.close();
					FileToWrite.add(outFile);
				} catch (Exception e)
				{
				}
			} else
			{
				break;
			}
		}

		FileToRead = FileToWrite;
		FileToWrite = new Vector<String>();

		while (FileToRead.size() > 0)
		{
			BlockForRead = new Block(BufferNumForsort - 1);
			blockois = new Vector<ObjectInputStream>();
			oos = null;

			outFile = GenerateFile();
			try
			{
				FileOutputStream fos = new FileOutputStream(outFile);
				oos = new ObjectOutputStream(fos);
			} catch (Exception e)
			{
			}
			
			int num;
			if((BufferNumForsort - 1)>FileToRead.size())
				num = FileToRead.size();
			else
				num = BufferNumForsort - 1;

			for (int i = 0; i < num; i++)
			{
				try
				{
					System.out.println("filename: " + FileToRead.elementAt(0));
					FileInputStream fis = new FileInputStream(FileToRead.elementAt(0));
					ObjectInputStream ois = new ObjectInputStream(fis);
					blockois.add(ois);

					BlockForRead.add(i, (Batch) ois.readObject());
					FileToRead.remove(0);
				} catch (Exception e)
				{
					FileToRead.remove(0);
					//System.out.println("error" + e.toString());
				}
			}

			while (BlockForRead.existTupleInBlock > 0)
			{
				min = BlockForRead.findMinFromSortedBatch(AttriIndex.firstElement());
				outBatch.add(BlockForRead.batchAt(min).elementAt(0));
				BlockForRead.batchAt(min).remove(0);
				BlockForRead.existTupleInBlock--;

				if (BlockForRead.batchAt(min).size() == 0)
				{
					BlockForRead.remove(min);
					try
					{
						Batch newBatch = (Batch) blockois.elementAt(min).readObject();
						BlockForRead.add(min, newBatch);
					} catch (EOFException e)
					{
						try
						{
							blockois.elementAt(min).close();
							blockois.remove(min);
						} catch (Exception e1)
						{
						}
					} catch (Exception e)
					{
					}
				}

				if (outBatch.isFull())
				{
					oos.writeObject(outBatch);
					IOCost++;
					outBatch = new Batch(batchSize);
				}
			}
			if (outBatch.size() > 0)
			{
				oos.writeObject(outBatch);
				IOCost++;
				outBatch = new Batch(batchSize);
			}

			FileToWrite.add(outFile);
			if (FileToRead.size() == 0)
			{
				if (FileToWrite.size() == 1)
				{
					System.out.println("Stage@ "+stage+" IOCost in BJ is "+IOCost);
					stage++;		
					return FileToWrite.firstElement();
				} else
				{
					FileToRead = FileToWrite;
					FileToWrite = new Vector<String>();
				}
			}
		}
		return null;
	}
}
